package concurrent.executors;

import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import utils.Log;

/**
 * 实现了CompletionService，将执行完成的任务放到阻塞队列中，通过take或poll方法来获得执行结果
 */
public class CompletionServiceTest {

    public static void main(String[] args) {
        try {
            count1();
        } catch (Throwable tr) {
            tr.printStackTrace();
        }
    }

    void test() throws InterruptedException, ExecutionException {
        ExecutorService executor = Executors.newFixedThreadPool(10);        //创建含10.条线程的线程池
        CompletionService completionService = new ExecutorCompletionService(executor);
        for (int i = 1; i <= 10; i++) {
            final int result = i;
            completionService.submit(new Callable() {
                public Object call() throws Exception {
                    Thread.sleep(new Random().nextInt(10000));   //让当前线程随机休眠一段时间
                    Log.vn("result:" + result);
                    return result;
                }
            });
        }
        System.out.println(completionService.take().get());   //获取执行结果
        Log.vn("=======over?====");
    }


    /***
     * 使用阻塞容器保存每次Executor处理的结果，在后面进行统一处理
     * @throws Exception

     */

    public static void count1() throws Exception {

        ExecutorService exec = Executors.newCachedThreadPool();

        BlockingQueue<Future<Integer>> queue = new LinkedBlockingQueue<Future<Integer>>();

        for (int i = 0; i < 100; i++) {

            Future<Integer> future = exec.submit(getTask());
            queue.add(future);

        }

        int sum = 0;

        int queueSize = queue.size();

        for (int i = 0; i < queueSize; i++) {

            sum += queue.take().get();

        }

        System.out.println("\n总数为：" + sum);

        exec.shutdown();

    }

    /***
     * 使用CompletionService(完成服务)保持Executor处理的结果
     * @throws InterruptedException

     * @throws ExecutionException

     */

    public static void count2() throws InterruptedException, ExecutionException {

        ExecutorService exec = Executors.newCachedThreadPool();

        CompletionService<Integer> execcomp = new ExecutorCompletionService<Integer>(exec);

        for (int i = 0; i < 10; i++) {

            execcomp.submit(getTask());

        }

        int sum = 0;

        for (int i = 0; i < 10; i++) {

            //检索并移除表示下一个已完成任务的 Future，如果目前不存在这样的任务，则等待。

            Future<Integer> future = execcomp.take();

            sum += future.get();

        }

        System.out.println("总数为：" + sum);

        exec.shutdown();

    }

    /**
     * 得到一个任务
     *
     * @return Callable
     */

    public static Callable<Integer> getTask() {

        final Random rand = new Random();

        Callable<Integer> task = new Callable<Integer>() {

            @Override

            public Integer call() throws Exception {

                int i = rand.nextInt(10);

                int j = rand.nextInt(10);

                int sum = i * j;

                System.out.print(sum + "\n");
                Thread.sleep(new Random().nextInt(10000));   //让当前线程随机休眠一段时间

                return sum;

            }

        };

        return task;

    }
}
